import traceback
import json
from datetime import datetime

from air_web.config.config import config
from air_web.dw.common_fun import create_topic
from air_web.web_flask.bll.base_bll import BaseBll
from air_web.web_flask.common.constants import Status, StatusMsg


class NotifyBll(BaseBll):
    C_CONS = config["C_CONS"]

    def __init__(self):
        """
        如果当前页面用到mysql则init_mysql_dal=True，self.sql_dal就会提供mysql实例，不用到则传False
        如果当前页面用到es则init_es_dal=True，self.es_dal就会提供es实例，不用到则传False
        """
        super().__init__(init_mysql_dal=True, init_kafka_producer=True)

    def control_res_motify(self, compute_date=None, on5=None):
        try:
            org_no = on5  # 地市或者区县

            sql = f"""select org_no
                      from city_approval_info
                      where control_time <= '{compute_date}' 
                        and end_time >= '{compute_date}'
                        and org_no = {org_no}
                   """
            approval_df = self.sql_dal.sql_pandas.query(sql)
            if len(str(on5)) == 5:
                where_org_field = 'on5'
                if approval_df.empty:  # 区县计划,返回区县
                    select_org_field = 'on7'
                    on_org_field = 'on7'
                else:  # 地市计划,返回地市
                    select_org_field = 'on5'
                    on_org_field = 'on5'
            else:
                select_org_field = 'on7'
                where_org_field = 'on7'
                if approval_df.empty:  # 地市计划,返回区县
                    on_org_field = 'on5'
                else:  # 区县计划,返回区县
                    on_org_field = 'on7'

            sql = f"""select distinct plan_id as planId, cc.{select_org_field} as executeCityRange
                      from city_control cc
                      join city_approval_info cai 
                         on cc.{on_org_field}=cai.org_no and cc.control_time=cai.control_time
                      where compute_date = '{compute_date}' 
                        and {where_org_field}={org_no}
                        and n_cons_subst > 0 
                   """
            res_df = self.sql_dal.sql_pandas.query(sql)
            if res_df.empty:
                result = {"status": Status.EMPTY_DATA_ERROR, 'msg': '当前没有未达标地区,不能发送通知'}
                return result

            for col in ['planId', 'executeCityRange']:
                res_df[col] = res_df[col].astype(str)
            res_df['type'] = '1'  # 短信通知
            send_list = res_df.to_dict("record")
            
            topic_name = config['TOPIC_AIR_CONDITION_NOTICE']
            create_topic(topic_name)
            for send_dict in send_list:
                message = {'data': send_dict, 'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
                value = json.dumps(message).encode('utf-8')
                self.kafka_producer.send(topic_name, value=value)
            self.logger.info(f"成功写入kafka数据条数:{len(send_list)}")
            # 刷新并关闭Kafka生产者
            self.kafka_producer.flush()
            self.kafka_producer.close()

            result = {"status": Status.OK, 'msg': '通知成功'}
        except:
            self.logger.error(traceback.format_exc())
            result = {"status": Status.SERVER_ERROR, 'msg': '通知失败'}
        finally:
            return result
